package com.hx.listener;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.util.ObjectUtil;
import com.alibaba.excel.EasyExcel;
import com.alibaba.excel.context.AnalysisContext;
import com.alibaba.excel.event.AnalysisEventListener;
import com.hx.entry.CangKuImportDTO;
import com.hx.service.ICangKuService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.dao.DataAccessException;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.atomic.AtomicLong;

/**
 * 增强型仓库数据导入监听器（用于EasyExcel处理百万级数据导入）
 * 功能：逐行读取Excel数据 → 校验 → 批量入库 → 错误处理 → 生成错误报告
 */
@Slf4j
public class EnhancedCangKuListener extends AnalysisEventListener<CangKuImportDTO> {
    // 动态批次大小（默认初始值50000条，根据处理情况自动调整）
    private int batchSize = 50000;

    // 数据缓存列表（存储当前批次待处理的合法数据）
    private List<CangKuImportDTO> cachedList = new ArrayList<>(batchSize);

    // 仓库服务接口（用于调用数据库批量插入操作）
    private final ICangKuService cangKuService;
    private ThreadPoolTaskExecutor executor;

    // 原子计数器（保证多线程安全）：
    private final AtomicLong totalCounter = new AtomicLong(0);   // 总处理数据量统计
    private final AtomicLong successCounter = new AtomicLong(0); // 成功入库数据量统计

    // 错误记录列表（线程安全的同步列表，存储所有校验失败或插入失败的数据）
    private final List<CangKuImportDTO> errorRecords = Collections.synchronizedList(new ArrayList<>());

    public EnhancedCangKuListener(ICangKuService cangKuService) {
        this.cangKuService = cangKuService;
        this.executor = threadPoolTaskExecutor();
    }

    /**
     * 核心处理方法：每读取一行Excel数据时触发
     *
     * @param data    当前行的数据对象
     * @param context EasyExcel解析上下文（包含工作表、行号等信息）
     */
    @Override
    public void invoke(CangKuImportDTO data, AnalysisContext context) {
        if (validateData(data)) {
            cachedList.add(data);
            //当缓存数据量达到批次大小时触发批量插入
            if (cachedList.size() >= batchSize) {
                List<CangKuImportDTO> sub = CollUtil.sub(cachedList, 0, batchSize);
                cachedList.clear();
                CompletableFuture.supplyAsync(() -> {
                    Long code = sub.stream().findFirst().get().getCode();
                    log.info("{}开始了", code);
                    try {
                        cangKuService.saveBatch(sub);
                        // 成功计数累加
                        successCounter.addAndGet(sub.size());
                    } catch (DataAccessException e) {
                        log.error("批次插入失败，失败数量：{}", sub.size(), e);
                        errorRecords.addAll(sub);
                    }
                    //清空缓存列表（无论成功与否）
                    sub.clear();
                    return code + "入库成功了";
                }, executor).thenAccept(info -> log.info("{}", info));

                adjustBatchSize(); // 动态调整批次大小
            }
        } else {
            errorRecords.add(data);
        }
        //定期打印处理进度（每处理1万条打印一次）
        //printProgress();
    }

    private boolean validateData(CangKuImportDTO data) {
        // 示例校验逻辑
        return ObjectUtil.isNotNull(data.getCode());
        //&& data.getAreaZD() != null
        //&& data.getCangKuDate() != null;
    }

    /**
     * 批量保存数据到数据库（核心插入逻辑）
     */
    private void saveBatch() {
        if (!cachedList.isEmpty()) {
            try {
                //cangKuService.batchInsert(cachedList);
                cangKuService.saveBatch(cachedList);
                // 成功计数累加
                successCounter.addAndGet(cachedList.size());
            } catch (DataAccessException e) {
                log.error("批次插入失败，失败数量：{}", cachedList.size(), e);
                errorRecords.addAll(cachedList);
            }
            //清空缓存列表（无论成功与否）
            cachedList.clear();
        }
    }

    /**
     * 动态调整批次大小（示例算法：根据处理情况自动优化）
     * 说明：如果批次过小会增加数据库交互次数，过大会导致内存压力
     */
    private void adjustBatchSize() {
        // 动态调整逻辑（示例）
        if (batchSize < 50000) {
            batchSize = Math.min(batchSize * 2, 50000);
        }
    }

    private void printProgress() {
        long total = totalCounter.incrementAndGet();
        if (total % 10000 == 0) {
            log.info("已处理：{} 条，成功：{} 条，失败：{} 条", total, successCounter.get(), errorRecords.size());
        }
    }

    /**
     * 所有数据解析完成后触发的收尾操作
     */
    @Override
    public void doAfterAllAnalysed(AnalysisContext context) {
        saveBatch(); // 处理最后一批数据
        //generateErrorReport(); // 生成错误报告
        log.info("导入完成！总计：{} 条，成功：{} 条，失败：{} 条", totalCounter.get(), successCounter.get(), errorRecords.size());
    }

    private void generateErrorReport() {
        if (!errorRecords.isEmpty()) {
            String errorFilePath = "/tmp/import_errors_" + System.currentTimeMillis() + ".xlsx";
            EasyExcel.write(errorFilePath, CangKuImportDTO.class)
                    .sheet("sheet1")
                    .doWrite(errorRecords);
            log.warn("错误数据已导出至：{}", errorFilePath);
        }
    }

    private ThreadPoolTaskExecutor threadPoolTaskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        int i = Runtime.getRuntime().availableProcessors();
        //核心线程数目
        executor.setCorePoolSize(2);
        //指定最大线程数
        executor.setMaxPoolSize(4);
        //队列中最大的数目
        executor.setQueueCapacity(2);
        //线程名称前缀
        executor.setThreadNamePrefix("ThreadPoolTaskExecutor-");
        //rejection-policy：当pool已经达到max size的时候，如何处理新任务
        //CALLER_RUNS：不在新线程中执行任务，而是由调用者所在的线程来执行
        //对拒绝task的处理策略
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        //executor.setRejectedExecutionHandler(new ThreadPoolExecutor.DiscardOldestPolicy());
        //executor.setRejectedExecutionHandler(new ThreadPoolExecutor.DiscardPolicy());
        //executor.setRejectedExecutionHandler(new ThreadPoolExecutor.AbortPolicy());
        //当调度器shutdown被调用时等待当前被调度的任务完成
        executor.setWaitForTasksToCompleteOnShutdown(true);
        //线程空闲后的最大存活时间
        executor.setKeepAliveSeconds(60);
        //加载
        executor.initialize();
        log.info("初始化线程池成功");
        return executor;
    }
}